package program

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Student {

  def main(args: Array[String]): Unit = {
    // 创建 Spark 配置对象，设置应用名称和运行模式（本地模式）
    val conf = new SparkConf().setAppName("GenderCount").setMaster("local[*]")
    // 创建 Spark Streaming 上下文，批处理间隔为 2 秒
    val ssc = new StreamingContext(conf, Seconds(2))

    // 定义 Kafka 参数，包括服务器地址、序列化器、消费者组 ID、偏移量重置策略等
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "8.152.219.6:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "niit008",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // 定义要订阅的 Kafka 主题
    val topics = Array("stuInfoTest")
    // 创建 Kafka Direct Stream，直接从 Kafka 读取数据
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    // 创建累加器，用于统计男性和女性的总数
    val totalMales = ssc.sparkContext.longAccumulator("TotalMales")
    val totalFemales = ssc.sparkContext.longAccumulator("TotalFemales")


    // 对每个 RDD 应用操作
    stream.foreachRDD { rdd =>
      // 如果 RDD 不为空，则进行处理
      if (!rdd.isEmpty()) {
        // 遍历 RDD 中的每个记录
        rdd.foreach { record =>
          // 解析记录中的值，假设记录格式为 "其他字段 性别"
          val fields = record.value().split("\\s+|\\t")
          // 如果字段长度至少为 3，则处理性别字段
          if (fields.length >= 3) {
            val gender = fields(2).toInt
            // 根据性别更新累加器
            if (gender == 1) totalMales.add(1)
            else if (gender == 0) totalFemales.add(1)
          }
        }
        // 打印到目前为止累计的男性和女性总数
        println(s"Total males so far: ${totalMales.value}")
        println(s"Total females so far: ${totalFemales.value}")
      }
    }
    // 启动 Spark Streaming 上下文
    ssc.start()
    // 等待流计算终止
    ssc.awaitTermination()
  }
}
